package com.example.reactor.demo.backpressure;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

/**
 * @author zpMeng
 * @since 9/22/2025
 */
public class TestBackPressureByFlux {

    public static void main(String[] args) throws InterruptedException {
        try {
            method1();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 方法1: 手动控制请求 + 慢处理
     */
    private static void method1() throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);

        Flux
                .fromStream(IntStream.rangeClosed(1, 100).boxed())
                .flatMap(Flux::just)
                .log("generate")
                .publishOn(Schedulers.boundedElastic(), 256)
                .log("after publish on")
                .bufferUntil(i -> i % 2 == 0 )
                .log("after buffer")
                .subscribe(new Subscriber<>() {
                    private Subscription subscription;

                    @Override
                    public void onSubscribe(Subscription s) {
                        this.subscription = s;
                        System.out.println("初始请求: 2");
                        s.request(3); // 关键：只请求1个！
                    }

                    @Override
                    public void onNext(List<Integer> buffer) {
                        System.out.println("处理缓冲区， 数据: " + buffer);
                        subscription.request(1); // 关键：只请求1个！

                        // 故意不调用 request() 请求更多数据
                        // 这样 bufferUntil 准备好发射下一个缓冲区时就没有请求了
                    }

                    @Override
                    public void onError(Throwable t) {
                        System.err.println("✅ 错误: " + t.getMessage());
                        latch.countDown();
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("完成");
                        latch.countDown();
                    }
                });

        latch.await();
    }


}
